package com.athuahua.rector;

import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;
import java.util.List;

/**
 * @Author: huahua
 * @Date: 2024/10/29 0029
 * @Time: 15:17
 * @Description:
 */
public class FluxDemo {

    //flux  N个元素的流
    @Test
    public void fluxTest01() throws Exception {
        // Flux : N个元素的流
        //发布者发布数据流：源头

        //1.多元素的流
        // 流的特性：流不消费就没用,流可以有无数个消费者[广播模式]
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);

        //消费：订阅
        just.subscribe(e -> System.out.println("e1= " + e));
        just.subscribe(e -> System.out.println("e2= " + e));

        System.out.println("-------------");
        //定时消费
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1));//每秒参数一个递增数据

        flux.subscribe(System.out::println);


        Thread.sleep(100000);


    }

    //Mnon 1/0个元素的流
    @Test
    public void MonoTest01() {
        //Mono<Integer> 只有一个Integer
        //Flux<Integer> 有多个Integer
        Mono<Integer> just = Mono.just(1);

        just.subscribe(System.out::println);
    }

    //事件感知API,当流发生什么事的时，触发一个回调（Hook[钩子函数]） : doOnxxx
    //doOnxxx要感知某个流的事件，需要写在这个流的后面，新流的前面
    @Test
    public void doOnXXXTest() throws Exception {
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5, 6).delayElements(Duration.ofSeconds(1)) //调频 每隔一秒发送一个元素
                .doOnComplete(() -> {
                    System.out.println("流结束了");
                }).doOnError(throwable -> { //感知流本身的异常
                    System.out.println("流出错了" + throwable);
                }).doOnNext(integer -> {
                    System.out.println("流获得下一个元素" + integer);
                });


        just.subscribe(value -> {
//            if (value < 5)
//                System.out.println(value);
//            else{
//                int i = 10/0;
//            }
            System.out.println(value);
        });


        Thread.sleep(2);


        System.in.read();

    }

    @Test
    public void test01() {
        Flux.range(1, 7)
                //     .log() //日志
                .filter(i -> i > 3).log().map(i -> "haha" + i).subscribe(System.out::println);


    }

    //subscribe 订阅流的方法
    @Test
    public void subScript() {

        Flux<String> flux = Flux.range(1, 10)
                .map(i -> {
                    //System.out.println("map...." + i);
//                    if (i == 9)
//                        i = 10 / (9 - i); //数学运算异常
                    return "哈哈" + i;
                });


        //流被订阅时才会流动
        //传递一个消费者
        //flux.subscribe(v -> System.out.println("v = " + v));

        //传递一个消费者 + 错误的消费者  + 完成时
//        flux.subscribe(
//                v -> System.out.println("v = " + v),       //流元素消费
//                error -> System.out.println(error),        //感知异常结束
//                () -> System.out.println("流结束了")         //感知正常结束
//
//        );


        //流的生命周期钩子可以传播给订阅者
        flux.subscribe(new BaseSubscriber<String>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                //流被订阅触发
                System.out.println("绑定了..." + subscription);
                subscription.request(1);
                // requestUnbounded(); //要无限数据
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("感知到数据到达： " + value);
                request(1);
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束了");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常..." + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消了、、、");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("无论正常还是异常都会执行" + type);
            }
        });


    }


    //buffer 缓冲区 ：消费一次最多可以拿到N个元素
    @Test
    public void bufferTest() {
        Flux<List<Integer>> buffer = Flux.range(1, 10)
                .buffer(3);//缓冲区


        /**
         * 类型: class java.util.ArrayList 值: [1, 2, 3]
         * 类型: class java.util.ArrayList 值: [4, 5, 6]
         * 类型: class java.util.ArrayList 值: [7, 8, 9]
         * 类型: class java.util.ArrayList 值: [10]
         * 凑满数批量发给消费者
         */
        buffer.subscribe(v -> System.out.println("类型: " + v.getClass() + " 值: " + v));


    }

    //limit 限流 ： 一次取N个元素
    @Test
    public void limitTest() {
        Flux.range(1, 1000)
                .log()
                .limitRate(100) //一次预取30个元素
                .subscribe();
        // 75% 预取策略
        // 第一次抓取100个数据，如果75%的元素已经被处理了，继续抓取新的 75%的元素


    }

    //generate 编程方式创建序列
    //Sink: 接收器
    //Source 源
    @Test
    public void generateTest() {
        Flux<Object> generate = Flux.generate(
                //初始值
                () -> 0,
                (state, sink) -> {//可能会抛出运行时异常【不受检异常】
                    //从0 - 10
                    if (state <= 10) {
                        sink.next(state); //把元素传出去
                    } else {
                        sink.complete(); //完成
                    }
                    return state + 1;
                });

        generate.log().subscribe();


    }

    //create 编程方式创建序列 ： 异步/多线程 创建流
    @Test
    public void createTest() {
        Flux.create(fluxSink -> {
            fluxSink.next("哈哈");
        });


    }

    //dispose 在其他线程中取消流
    @Test
    public void disposeTest() throws Exception {
        Flux<Integer> log = Flux.range(1, 10000)
                .delayElements(Duration.ofSeconds(1))
                .map(i -> i + 7)
                .log();

        //可取消的
        Disposable subscribe = log.subscribe(System.out::println);


        new Thread(() -> {
            try {
                Thread.sleep(10000);
                subscribe.dispose(); //销毁
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        ).start();


        System.in.read();

    }

    //handle 自定义流中元素处理规则 比map更灵活
    @Test
    public void handleTest() {
        Flux.range(1, 10)
                .handle((value, sink) -> { //接受 元素 和 通道
                    //System.out.println("值：" + value);
                    String string = "值：" + value.toString();//处理元素

                    sink.next(string); //向下发送数据的通道
                })
                .log()
                .subscribe();


    }


}
